package org.jetlinks.pro.things.data;

import com.alibaba.fastjson.JSON;
import com.google.common.collect.Maps;
import io.vavr.Function3;
import io.vavr.Function4;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.codec.digest.DigestUtils;
import org.apache.commons.collections.MapUtils;
import org.hswebframework.web.api.crud.entity.PagerResult;
import org.hswebframework.web.api.crud.entity.QueryParamEntity;
import org.hswebframework.web.exception.BusinessException;
import org.hswebframework.web.id.IDGenerator;
import org.jetlinks.core.message.DeviceLogMessage;
import org.jetlinks.core.message.Headers;
import org.jetlinks.core.message.ThingMessage;
import org.jetlinks.core.message.event.EventMessage;
import org.jetlinks.core.message.event.ThingEventMessage;
import org.jetlinks.core.message.property.PropertyMessage;
import org.jetlinks.core.metadata.*;
import org.jetlinks.core.metadata.types.*;
import org.jetlinks.core.things.ThingMetadata;
import org.jetlinks.core.things.ThingTemplate;
import org.jetlinks.core.things.ThingType;
import org.jetlinks.core.things.ThingsRegistry;
import org.jetlinks.core.utils.DeviceMessageTracer;
import org.jetlinks.core.utils.TimestampUtils;
import org.jetlinks.pro.gateway.DeviceMessageUtils;
import org.jetlinks.pro.timeseries.TimeSeriesData;
import org.jetlinks.pro.timeseries.query.AggregationData;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.util.function.Tuple2;
import reactor.util.function.Tuples;

import javax.annotation.Nonnull;
import javax.annotation.Nullable;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Function;
import java.util.stream.Collectors;

@AllArgsConstructor
@Slf4j
public abstract class AbstractThingDataStorageStrategy implements ThingDataStorageStrategy {

    protected final ThingsRegistry registry;

    /**
     * 执行保存单个数据
     *
     * @param table 指标ID
     * @param data  数据
     * @return void
     */
    protected abstract Mono<Void> doSaveData(String table, TimeSeriesData data);


    protected abstract Flux<Tuple2<String, TimeSeriesData>> convertProperties(String templateId,
                                                                              ThingMessage message,
                                                                              Map<String, Object> properties,
                                                                              Map<String, Long> propertySourceTimes);

    protected abstract <T> Flux<T> doQuery(String table,
                                           QueryParamEntity paramEntity,
                                           Function<TimeSeriesData, T> mapper);

    protected abstract <T> Mono<PagerResult<T>> doQueryPager(String table,
                                                             QueryParamEntity paramEntity,
                                                             Function<TimeSeriesData, T> mapper);


    protected String createDataId(ThingMessage message) {
        long ts = message.getTimestamp();
        return DigestUtils.md5Hex(String.join("_", message.getThingId(), String.valueOf(createUniqueNanoTime(ts))));
    }

    protected Mono<Tuple2<String, TimeSeriesData>> createDeviceMessageLog(String templateId,
                                                                          ThingMessage message) {


        return Mono.just(Tuples.of(getMessageLogTable(message.getThingType(), templateId, message.getThingId()),
                                   TimeSeriesData.of(message.getTimestamp(), createDeviceLogData(templateId, message))));
    }

    protected abstract String getMessageLogTable(String thingType, String templateId, String thingId);

    protected abstract String getEventTable(String thingType, String templateId, String thingId, String eventId);

    protected abstract String getPropertyTable(String thingType, String productId, String thingId);

    protected Map<String, Object> createDeviceLogData(String templateId, ThingMessage message) {
        Map<String, Object> data = Maps.newHashMapWithExpectedSize(8);
        data.put("id", IDGenerator.SNOW_FLAKE_STRING.generate());
        data.put(thingIdColumn(), message.getThingId());
        data.put("timestamp", message.getTimestamp());
        data.put("createTime", System.currentTimeMillis());
        data.put("messageId", message.getMessageId());
        data.put("type", message.getMessageType().name());
        String log;
        if (message instanceof DeviceLogMessage) {
            log = ((DeviceLogMessage) message).getLog();
        } else {
            log = message.getHeader("log").map(String::valueOf).orElseGet(() -> message.toJson().toJSONString());
        }
        data.put("content", log);
        return data;
    }

    private Flux<Tuple2<String, TimeSeriesData>> convertMessageToTimeSeriesData(ThingMessage message) {
        boolean ignoreStorage = message.getHeaderOrDefault(Headers.ignoreStorage);
        boolean ignoreLog = message.getHeaderOrDefault(Headers.ignoreLog);
        if (ignoreStorage && ignoreLog) {
            return Flux.empty();
        }
        DeviceMessageTracer.trace(message, "save.before");
        String templateId = (String) message.getHeader("templateId").orElse("null");
        List<Publisher<Tuple2<String, TimeSeriesData>>> all = new ArrayList<>(2);
        //没有忽略数据存储
        if (!ignoreStorage) {
            //事件上报
            if (message instanceof EventMessage) {
                all.add(convertEventMessageToTimeSeriesData(templateId, ((EventMessage) message)));
            } else {
                //属性相关
                Map<String, Object> properties = DeviceMessageUtils
                    .tryGetProperties(message)
                    .orElseGet(Collections::emptyMap);
                //属性源时间
                Map<String, Long> propertiesTimes = DeviceMessageUtils
                    .tryGetPropertySourceTimes(message)
                    .orElseGet(Collections::emptyMap);
                if (MapUtils.isNotEmpty(properties)) {
                    all.add(convertProperties(templateId, message, properties, propertiesTimes));
                }
            }
        }
        if (!ignoreLog) {
            all.add(createDeviceMessageLog(templateId, message));
        }
        return Flux.merge(all);
    }

    protected Mono<Tuple2<String, TimeSeriesData>> convertEventMessageToTimeSeriesData(String templateId,
                                                                                       ThingEventMessage message) {

        return registry
            .getTemplate(message.getThingType(), templateId)
            .flatMap(template -> template
                .getMetadata()
                .<TimeSeriesData>handle((metadata, sink) -> {
                    if (metadata.getEventOrNull(message.getEvent()) == null) {
                        log.warn("物模版[{}]未定义事件物模型:{}", templateId, message.getEvent());
                        return;
                    }
                    Map<String, Object> data = createEventData(message, metadata);
                    sink.next(TimeSeriesData.of(TimestampUtils.toMillis(message.getTimestamp()), data));
                }))
            .map(data -> Tuples.of(getEventTable(message.getThingType(), templateId, message.getThingId(), message.getEvent()), data));
    }

    protected Map<String, Object> createEventData(ThingEventMessage message, ThingMetadata metadata) {
        Object value = message.getData();
        DataType dataType = metadata
            .getEvent(message.getEvent())
            .map(EventMetadata::getType)
            .orElseGet(UnknownType::new);
        Object tempValue = dataType instanceof Converter ? ((Converter<?>) dataType).convert(value) : value;
        Map<String, Object> data;
        if (tempValue instanceof Map) {
            @SuppressWarnings("all")
            Map<String, Object> mapValue = ((Map) tempValue);
            int size = mapValue.size();
            data = newMap(size);
            data.putAll(mapValue);
            //只记录物模型中记录的字段
            if (dataType instanceof ObjectType) {
                Set<String> nonexistent = new HashSet<>(data.keySet());
                ObjectType objType = ((ObjectType) dataType);
                for (PropertyMetadata property : objType.getProperties()) {
                    nonexistent.remove(property.getId());
                }
                nonexistent.forEach(data::remove);
            }
        } else {
            data = newMap(16);
            data.put("value", tempValue);
        }
        data.put("id", createDataId(message));
        data.put(thingIdColumn(), message.getThingId());
        data.put("createTime", System.currentTimeMillis());
        data.put("timestamp", message.getTimestamp());

        return data;
    }

    protected final Flux<ThingPropertyDetail> rowToProperty(TimeSeriesData row, Collection<PropertyMetadata> properties) {
        return Flux.fromIterable(rowToPropertyList(row, properties));
    }

    protected final List<ThingPropertyDetail> rowToPropertyList(TimeSeriesData row, Collection<PropertyMetadata> properties) {
        return properties
            .stream()
            .filter(prop -> row.get(prop.getId()).isPresent())
            .map(property -> ThingPropertyDetail
                .of(row,
                    row.get(property.getId()).orElse(0),
                    property
                ).property(property.getId()))
            .collect(Collectors.toList());
    }

    protected boolean propertyIsJsonStringStorage(PropertyMetadata metadata) {

        return false;
    }

    protected boolean propertyIsIgnoreStorage(PropertyMetadata metadata) {

        return false;
    }

    protected Object convertPropertyValue(Object value, PropertyMetadata metadata) {
        if (value == null || metadata == null) {
            return value;
        }
        //使用json字符串来存储
        if (propertyIsJsonStringStorage(metadata)) {
            return value instanceof String ? String.valueOf(value) : JSON.toJSONString(value);
        }
        //数字类型直接返回
        if (metadata.getValueType() instanceof NumberType && value instanceof Number) {
            return value;
        }
        if (metadata.getValueType() instanceof Converter) {
            return ((Converter<?>) metadata.getValueType()).convert(value);
        }
        return value;
    }

    protected Flux<Tuple2<String, TimeSeriesData>> convertPropertiesForColumnPolicy(String templateId,
                                                                                    ThingMessage message,
                                                                                    Map<String, Object> properties) {
        if (MapUtils.isEmpty(properties)) {
            return Flux.empty();
        }
        return this
            .registry
            .getThing(ThingType.of(message.getThingType()), message.getThingId())
            .flatMapMany(thing -> thing
                .getMetadata()
                .flatMap(metadata -> {
                    int size = properties.size();
                    String id;
                    //强制使用时间戳作为数据ID
                    if (message.getHeader(Headers.useTimestampAsId).orElse(false)) {
                        id = String.join("_", message.getThingId(), String.valueOf(message.getTimestamp()));
                    } else {
                        id = createDataId(message);
                    }
                    Map<String, Object> data = newMap(size);
                    //转换属性数据
                    for (Map.Entry<String, Object> entry : properties.entrySet()) {
                        PropertyMetadata propertyMetadata = metadata.getPropertyOrNull(entry.getKey());
                        //没有配置物模型或者忽略了存储
                        if (propertyMetadata == null || propertyIsIgnoreStorage(propertyMetadata)) {
                            continue;
                        }
                        Object value = convertPropertyValue(entry.getValue(), propertyMetadata);
                        if (null != value) {
                            data.put(entry.getKey(), value);
                        }
                    }
                    //没有属性值,可能全部都配置了不存储
                    if (data.isEmpty()) {
                        return Mono.empty();
                    }
                    data.put(thingIdColumn(), message.getThingId());
                    data.put("timestamp", TimestampUtils.toMillis(message.getTimestamp()));
                    data.put("createTime", System.currentTimeMillis());
                    data.put("id", DigestUtils.md5Hex(id));

                    return Mono
                        .just(Tuples
                                  .of(getPropertyTable(thing.getType().getId(), templateId, thing.getId()),
                                      TimeSeriesData.of(message.getTimestamp(), data))
                        );
                }));
    }

    private Map<String, Object> newMap(int size) {
        return Maps.newHashMapWithExpectedSize(size);
    }

    protected Flux<Tuple2<String, TimeSeriesData>> convertPropertiesForRowPolicy(String templateId,
                                                                                 ThingMessage message,
                                                                                 Map<String, Object> properties) {
        if (org.apache.commons.collections4.MapUtils.isEmpty(properties)) {
            return Flux.empty();
        }
        return this
            .registry
            .getThing(ThingType.of(message.getThingType()), message.getThingId())
            .flatMapMany(thing -> thing
                .getMetadata()
                .flatMapIterable(metadata -> convertPropertiesToTimeSeriesData(message, metadata))
                .map(data -> Tuples.of(getPropertyTable(thing.getType().getId(), templateId, thing.getId()), data)));
    }

    private List<TimeSeriesData> convertPropertiesToTimeSeriesData(ThingMessage message, ThingMetadata metadata) {
        if (!(message instanceof PropertyMessage)) {
            return Collections.emptyList();
        }
        PropertyMessage propertyMessage = ((PropertyMessage) message);
        Map<String, Object> properties = propertyMessage.getProperties();
        if (org.apache.commons.collections4.MapUtils.isEmpty(properties)) {
            return Collections.emptyList();
        }

        List<TimeSeriesData> list = new ArrayList<>(properties.size());
        for (Map.Entry<String, Object> entry : properties.entrySet()) {
            String id;
            String property = entry.getKey();
            Object value = entry.getValue();
            if (value == null) {
                continue;
            }
            long ts = ((PropertyMessage) message).getPropertySourceTime(property).orElse(message.getTimestamp());
            //忽略存在没有的属性和忽略存储的属性
            PropertyMetadata propertyMetadata = metadata.getPropertyOrNull(property);
            if (propertyMetadata == null || propertyIsIgnoreStorage(propertyMetadata)) {
                continue;
            }
            //强制使用时间戳作为数据ID
            if (message.getHeader(Headers.useTimestampAsId).orElse(false)) {
                id = String.join("_", message.getThingId(), property, String.valueOf(message.getTimestamp()));
            } else {
                id = String.join("_", message.getThingId(), property, String.valueOf(createUniqueNanoTime(ts)));
            }
            list
                .add(TimeSeriesData.of(ts, this
                    .createRowPropertyData(id,
                                           TimestampUtils.toMillis(ts),
                                           message.getThingId(),
                                           propertyMetadata,
                                           entry.getValue()))
                );
        }

        return list;

    }

    protected Map<String, Object> createRowPropertyData(String id,
                                                        long timestamp,
                                                        String deviceId,
                                                        PropertyMetadata property,
                                                        Object value) {
        Map<String, Object> propertyData = newMap(24);
        propertyData.put("id", DigestUtils.md5Hex(id));
        propertyData.put("deviceId", deviceId);
        propertyData.put("timestamp", timestamp);
        propertyData.put("property", property.getId());
        propertyData.put("createTime", System.currentTimeMillis());

        fillRowPropertyValue(propertyData, property, value);
        return propertyData;
    }

    protected void fillRowPropertyValue(Map<String, Object> target, PropertyMetadata property, Object value) {
        if (value == null) {
            return;
        }
        if (property == null) {
            if (value instanceof Number) {
                target.put("numberValue", value);
            } else if (value instanceof Date) {
                target.put("timeValue", value);
            }
            target.put("value", String.valueOf(value));
            return;
        }
        DataType type = property.getValueType();
        target.put("type", type.getId());
        String convertedValue;
        if (type instanceof NumberType) {
            NumberType<?> numberType = (NumberType<?>) type;
            Number number = value instanceof Number
                ? ((Number) value)
                : numberType.convertNumber(value);
            if (number == null) {
                throw new BusinessException("error.cannot_convert", 500, value, type.getId());
            }
            convertedValue = String.valueOf(number);
            target.put("numberValue", number);
        } else if (type instanceof DateTimeType) {
            DateTimeType dateTimeType = (DateTimeType) type;
            convertedValue = String.valueOf(value);
            target.put("timeValue", dateTimeType.convert(convertedValue));
        } else if (propertyIsJsonStringStorage(property)) {
            //使用json字符来存储
            convertedValue = value instanceof String
                ? String.valueOf(value)
                : JSON.toJSONString(value);

        } else if (type instanceof ObjectType) {
            ObjectType objectType = (ObjectType) type;
            Object val = objectType.convert(value);
            convertedValue = JSON.toJSONString(val);
            target.put("objectValue", val);
        } else if (type instanceof ArrayType) {
            ArrayType objectType = (ArrayType) type;
            Object val = objectType.convert(value);
            convertedValue = JSON.toJSONString(val);
            target.put("arrayValue", val);
        } else if (type instanceof GeoType) {
            GeoType geoType = (GeoType) type;
            GeoPoint val = geoType.convert(value);
            convertedValue = String.valueOf(val);
            target.put("geoValue", val);
        } else {
            convertedValue = String.valueOf(value);
        }
        target.put("value", convertedValue);
    }

    protected Mono<Tuple2<ThingTemplate, ThingMetadata>> getTemplateMetadataByThing(String thingType, String thingId) {
        return registry
            .getThing(thingType, thingId)
            .flatMap(device -> Mono.zip(device.getTemplate(), device.getMetadata()));
    }

    protected Mono<Tuple2<ThingTemplate, ThingMetadata>> getTemplateMetadataByTemplate(String thingType, String templateId) {
        return registry
            .getTemplate(thingType, templateId)
            .zipWhen(ThingTemplate::getMetadata);
    }

    protected List<PropertyMetadata> getPropertyMetadata(DeviceMetadata metadata, String... properties) {
        if (properties == null || properties.length == 0) {
            return metadata.getProperties();
        }
        if (properties.length == 1) {
            return metadata.getProperty(properties[0])
                           .map(Arrays::asList)
                           .orElseGet(Collections::emptyList);
        }
        Set<String> ids = new HashSet<>(Arrays.asList(properties));
        return metadata
            .getProperties()
            .stream()
            .filter(prop -> ids.isEmpty() || ids.contains(prop.getId()))
            .collect(Collectors.toList());
    }


    private final AtomicInteger nanoInc = new AtomicInteger();

    protected long createUniqueNanoTime(long millis) {
        long nano = TimeUnit.MILLISECONDS.toNanos(millis);

        int inc = nanoInc.incrementAndGet();

        if (inc >= 99990) {
            nanoInc.set(inc = 1);
        }

        return nano + inc;
    }

    @Override
    public Mono<Void> save(ThingMessage message) {
        return this
            .convertMessageToTimeSeriesData(message)
            .flatMap(tp2 -> doSaveData(tp2.getT1(), tp2.getT2()))
            .then();
    }

    @Override
    public Flux<ThingPropertyDetail> queryProperty(@Nonnull String thingType,
                                                   @Nonnull String thingId,
                                                   @Nonnull String property,
                                                   @Nonnull QueryParamEntity param) {
        return null;
    }

    @Override
    public Flux<ThingPropertyDetail> queryProperties(@Nonnull String thingType,
                                                     @Nonnull String thingId,
                                                     @Nullable Collection<String> properties,
                                                     @Nonnull QueryParamEntity param) {
        return null;
    }

    @Override
    public Mono<PagerResult<ThingPropertyDetail>> queryPropertyPage(@Nonnull String thingType,
                                                                    @Nonnull String thingId,
                                                                    @Nonnull String property,
                                                                    @Nonnull QueryParamEntity param) {
        return null;
    }

    @Override
    public Flux<AggregationData> aggregationProperties(@Nonnull String thingType,
                                                       @Nonnull String thingId,
                                                       @Nonnull AggregationRequest request,
                                                       @Nonnull PropertyAggregation... properties) {
        return null;
    }

    @Override
    public Flux<AggregationData> aggregationEvent(@Nonnull String thingType,
                                                  @Nonnull String thingId,
                                                  @Nonnull String event,
                                                  @Nonnull AggregationRequest request,
                                                  @Nonnull PropertyAggregation... properties) {
        return null;
    }

    private String thingIdColumn() {
        return "thingId";
    }

    @Override
    public Flux<ThingEvent> queryEvent(@Nonnull String thingType,
                                       @Nonnull String thingId,
                                       @Nonnull String event,
                                       @Nonnull QueryParamEntity param) {
        return this
            .doQuery(thingType, thingId, param,
                     (_thingType, _templateId, _thingId) -> getEventTable(_thingType, _templateId, _thingId, event),
                     ThingEvent::of);
    }

    @Override
    public Mono<PagerResult<ThingEvent>> queryEventPage(@Nonnull String thingType,
                                                        @Nonnull String thingId,
                                                        @Nonnull String event,
                                                        @Nonnull QueryParamEntity param) {

        return this
            .doQueryPager(thingType, thingId, param,
                          (_thingType, _templateId, _thingId) -> getEventTable(_thingType, _templateId, _thingId, event),
                          ThingEvent::of);

    }

    @Override
    public Flux<ThingMessageLog> queryMessageLog(@Nonnull String thingType,
                                                 @Nonnull String thingId,
                                                 @Nonnull QueryParamEntity param) {
        return this
            .doQuery(thingType, thingId, param, this::getMessageLogTable, ThingMessageLog::of);
    }

    @Override
    public Mono<PagerResult<ThingMessageLog>> queryMessageLogPage(@Nonnull String thingType,
                                                                  @Nonnull String thingId,
                                                                  @Nonnull QueryParamEntity param) {

        return this
            .doQueryPager(thingType, thingId, param, this::getMessageLogTable, ThingMessageLog::of);
    }

    private <T> Flux<T> doQuery(String thingType,
                                String thingId,
                                QueryParamEntity param,
                                Function3<String, String, String, String> tableBuilder,
                                Function<TimeSeriesData, T> converter) {
        return this
            .getTemplateMetadataByThing(thingType, thingId)
            .flatMapMany(tp2 -> param
                .toQuery()
                .where(thingIdColumn(), thingId)
                .execute(queryParam -> this
                    .doQuery(tableBuilder.apply(thingType, tp2.getT1().getId(), thingId),
                             param,
                             converter)));
    }

    private <T> Mono<PagerResult<T>> doQueryPager(String thingType,
                                                  String thingId,
                                                  QueryParamEntity param,
                                                  Function3<String, String, String, String> tableBuilder,
                                                  Function<TimeSeriesData, T> converter) {
        return this
            .getTemplateMetadataByThing(thingType, thingId)
            .flatMap(tp2 -> param
                .toQuery()
                .where(thingIdColumn(), thingId)
                .execute(queryParam -> this
                    .doQueryPager(tableBuilder.apply(thingType, tp2.getT1().getId(), thingId),
                                  param,
                                  converter)));
    }
}
